package com.ultrapower.request;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import com.ultrapower.hbase.HbaseUtil;
import com.ultrapower.mapper.MapStatisticsCallerMapper;
import com.ultrapower.util.ApplicationMapStatisticsUtils;
import com.ultrapower.util.HBaseTables;
import com.ultrapower.util.TimeSlot;
import com.ultrapower.util.rowkey.RowKeyDistributorFactory;
import com.ultrapower.util.topology.LinkDataMap;
/**
 * 
 * @author wangy
 *
 */
public class ApplicationMapStatisticsCallee_Ver2Table extends BaseRequest {
	protected static final int MAP_STATISTICS_CALLEE_VER2_NUM_PARTITIONS = 32;
	private static final int SCAN_CACHE_SIZE = 40;
	private RowKeyDistributorByHashPrefix rowKeyDistributor = RowKeyDistributorFactory.createRowKeyDistributor(0, 36, 32);
	/**
	 * 
	 * @param args
	 */
	public static void main(String[] args) {
		BaseRequest table=new ApplicationMapStatisticsCallee_Ver2Table();
		table.execute();
	}

	@Override
	void execute() {
		TableName mapStatisticsCalleeTableName = HbaseUtil.getTableName(HBaseTables.MAP_STATISTICS_CALLEE_VER2_STR);
		MapStatisticsCallerMapper map = new MapStatisticsCallerMapper(rowKeyDistributor);
		this.getHbaseTemplate().find(mapStatisticsCalleeTableName, this.createScan(), rowKeyDistributor,
				(ResultScanner scanner) -> {
					scanner.forEach(result -> {
						try {
							LinkDataMap r = map.mapRow(result,0);
							System.out.println(r);
						} catch (Exception e) {
							e.printStackTrace();
						}

					});
					return null;
				});

	}

	@Override
	protected Scan createScan() {
		
		byte[] family = HBaseTables.MAP_STATISTICS_CALLEE_VER2_CF_COUNTER;
		TimeSlot timeslot = new TimeSlot();
		final long startTime = timeslot.getTimeSlot(from) - 1;
		final long endTime = timeslot.getTimeSlot(to);

		byte[] startKey = ApplicationMapStatisticsUtils.makeRowKey(applicationName, srviceTypeCode, endTime);
		byte[] endKey = ApplicationMapStatisticsUtils.makeRowKey(applicationName, srviceTypeCode, startTime);

		Scan scan = new Scan();
		scan.setCaching(SCAN_CACHE_SIZE);
		scan.setStartRow(startKey);
		scan.setStopRow(endKey);
		scan.addFamily(family);
		scan.setId("ApplicationStatisticsScan");

		return scan;
	}

}
